package org.myfram.flink.flinkonjar.application.job;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.myfram.flink.flinkonjar.common.config.KafkaConfig;
import org.myfram.flink.flinkonjar.common.connector.kafka.source.KafkaSourceBuilder;
import org.myfram.flink.flinkonjar.common.connector.oracle.sink.OracleSinkBuilder;
import org.myfram.flink.flinkonjar.common.properties.KafkaProperties;
import org.myfram.flink.flinkonjar.common.properties.OracleProperties;
import org.myfram.flink.flinkonjar.common.util.YamlUtils;
import org.myfram.flink.flinkonjar.domain.entity.KafkaToOracleJob.KafkaTargetUserEvent;
import org.myfram.flink.flinkonjar.domain.entity.KafkaToOracleJob.UserStatementBuilder;
import org.myfram.flink.flinkonjar.domain.process.UserFieldProcessFunction;

import java.util.Properties;

public class KafkaToOracleJob extends BaseFlinkJob {

    @Override
    protected void process(StreamExecutionEnvironment env) throws Exception {
        // 获取Kafka配置
        KafkaProperties kafkaProperties = YamlUtils.getObject("flink.kafka", KafkaProperties.class);
        OracleProperties oracleProperties = YamlUtils.getObject("flink.oracle", OracleProperties.class);
        Properties sourceProps = KafkaConfig.getSourceProps(kafkaProperties.getSource());

        // 构建Kafka Source
        DataStreamSource<String> kafkaSource = KafkaSourceBuilder.build(env, kafkaProperties.getSource().getTopic(), sourceProps);

        // 反序列化为实体
        SingleOutputStreamOperator<KafkaTargetUserEvent> kafkaStream = kafkaSource
                .map(new UserFieldProcessFunction());

        // Oracle Sink配置
        String url = "jdbc:oracle:thin:@//host:port/service";
        String username = "your_user";
        String password = "your_password";
        String insertSql = "INSERT INTO your_table (id, name) VALUES (?, ?)";

        kafkaStream.addSink(
                OracleSinkBuilder.build(
                        oracleProperties,
                        oracleProperties.getInsertSql(),
                        new UserStatementBuilder()
                )
        ).name("OracleSink");

        // 写入Oracle
        // kafkaStream.addSink(OracleSinkBuilder.build(url, username, password, insertSql));
    }
}